﻿using Misakai.Kafka;
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Linq;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using ZooKeeperNet;

namespace Reflexive.IO
{
    public class KafkaWriter<TConverter, TSource> : KafkaBase, IObserver<TSource> 
    where TConverter : ITypeConverter<TSource, byte[]>, new()
    {

        string _topic;
        string _endpoint;
        Producer _client;
        public StatusCode Status
        {
            get;
            private set;
        }

        public KafkaWriter(IPAddress ip, int port, string topic = null)
        {
            _topic = topic;
            _endpoint = String.Format("{0}:{1}", ip.ToString(), port.ToString());
            _client = new Producer(new BrokerRouter(GetOptions(_endpoint, Timeout.InfiniteTimeSpan)));
        }

        public void Stop()
        {
            Status = StatusCode.Suspended;
        }

        public void Start()
        {
            if (Status == StatusCode.Completed)
            {
                throw new NotSupportedException("Writer is in the Completed state.");
            }

            Status = StatusCode.Started;
        }

        public void OnCompleted()
        {
            if (Status != StatusCode.Completed)
            {
                Status = StatusCode.Completed;
                _client.Dispose();
            }
        }

        public void OnError(Exception e)
        {
            System.Diagnostics.Trace.TraceError(e.ToString());
        }

        public void OnNext(TSource value)
        {
            if (Status == StatusCode.Started)
            {
                try
                {
                    _client.SendMessageAsync(_topic, new[] { new Message() { Value = (Convertible<TConverter, TSource, byte[]>)value } }, 1, new TimeSpan(0, 0, 45));
                }
                catch
                {
                    lock (this)
                    {
                        _client = new Producer(new BrokerRouter(GetOptions(_endpoint, Timeout.InfiniteTimeSpan)));
                        _client.SendMessageAsync(_topic, new[] { new Message() { Value = (Convertible<TConverter, TSource, byte[]>)value } }, 1, new TimeSpan(0, 0, 45));
                    }

                }
            }
        }

        public void Dispose()
        {
            OnCompleted();
        }
    }
}
